package com.fmdj.snm.task;

import cn.hutool.core.collection.ListUtil;
import cn.hutool.core.map.MapUtil;
import com.fmdj.common.exception.GlobalException;
import com.fmdj.snm.entry.NewOrderMessage;
import com.rabbitmq.client.*;
import lombok.extern.slf4j.Slf4j;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;
import com.rabbitmq.client.Connection;
import com.rabbitmq.client.Channel;
import javax.annotation.Resource;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

@Component
@Slf4j
public class NewOrderMassageTask {
    @Resource
    private ConnectionFactory factory;

    /**
     * 同步发送消息: 交给当前线程完成
     * @param list
     */
    public void sendNewOrderMessage(ArrayList<NewOrderMessage> list) {
        int ttl = 1 * 60 * 1000; //设置新订单消息的过期时间为1分钟 1000毫秒=1秒
        String exchangeName = "new_order_private";
        try (
                //新特性:把释放资源交给JVM自动完成
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel()
        ) {
            //声明交换机
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);

            //队列可能存在的参数
            HashMap params = new HashMap<>();

            //遍历新订单消息进行封装
            for (NewOrderMessage newOrderMessage : list) {
                HashMap map = new HashMap<>(){{
                    put("orderId", newOrderMessage.getOrderId());
                    put("from", newOrderMessage.getFrom());
                    put("to", newOrderMessage.getTo());
                    put("expectsFee", newOrderMessage.getExpectsFee());
                    put("mileage", newOrderMessage.getMileage());
                    put("minute", newOrderMessage.getMinute());
                    put("distance", newOrderMessage.getDistance());
                    put("favourFee", newOrderMessage.getFavourFee());
                    put("userId", newOrderMessage.getUserId());
                }};

                // 构建属性参数
                AMQP.BasicProperties properties = new AMQP.BasicProperties()
                        .builder()
                        .contentEncoding("UTF-8")
                        .headers(map)
                        .expiration(ttl + "")
                        .build();

                //声明队列
                String queueName = "queue_"+ newOrderMessage.getUserId();
                String routingKey = newOrderMessage.getUserId();
                channel.queueDeclare(queueName, true, false, false, params);

                ////队列和交换机通过routingKey绑定
                channel.queueBind(queueName, exchangeName, routingKey);

                //发送消息
                channel.basicPublish(exchangeName,routingKey,properties,"发送新订单消息".getBytes());
                log.info(newOrderMessage.getUserId() + "发送新订单消息成功");
            }

        } catch (Exception e) {
            log.error("新订单消息发送失败", e);
            throw new GlobalException("新订单消息发送失败");
        }
    }

    /**
     * 异步发送消息
     * 例如: A方法(在主线程)调用该方法,不用等待sendNewOrderMessageAsync执行完,因为A方法和该方法已经在不同的线程中执行
     * @param list
     */
    @Async /*该注解代表异步执行*/
    public void sendNewOrderMessageAsync(ArrayList<NewOrderMessage> list) {
        this.sendNewOrderMessage(list); //异步方法中调用(本类)的同步方法 避免阻塞主线程
    }

    /**
     * 接收消息
     *
     * @param userId
     * @return
     */
    public List<NewOrderMessage> receiveNewOrderMessage(long userId) {
        String  exchangeName = "new_order_private";
        String queueName = "queue_" + userId;
        String routingKey = userId + "";
        try (
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel()
        ) {
            //这里交换机队列以及队列绑定做不做都行 防止另一端没做 把这一端也做了
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            channel.queueDeclare(queueName, true, false, false, null);
            channel.queueBind(queueName, exchangeName, routingKey);

            ArrayList<NewOrderMessage> list = new ArrayList<>();

            while (true) {
                //取出消息
                GetResponse response = channel.basicGet(queueName, false);//拒绝自动确认 自己手动确认
                channel.basicQos(0, 10, false); //批量处理消息 一次只取10条 防止内存溢出
                if (response!=null) {
                    AMQP.BasicProperties properties = response.getProps();
                    Map<String, Object> headers = properties.getHeaders();
                    String orderId = MapUtil.getStr(headers, "orderId");
                    String from = MapUtil.getStr(headers, "from");
                    String to = MapUtil.getStr(headers, "to");
                    String expectsFee = MapUtil.getStr(headers, "expectsFee");
                    String mileage = MapUtil.getStr(headers, "mileage");
                    String minute = MapUtil.getStr(headers, "minute");
                    String distance = MapUtil.getStr(headers, "distance");
                    String favourFee = MapUtil.getStr(headers, "favourFee");

                    //把消息内容封装进对象中
                    NewOrderMessage message = new NewOrderMessage();
                    message.setOrderId(orderId);
                    message.setFrom(from);
                    message.setTo(to);
                    message.setExpectsFee(expectsFee);
                    message.setMileage(mileage);
                    message.setMinute(minute);
                    message.setDistance(distance);
                    message.setFavourFee(favourFee);

                    //手动确认收到消息
                    long deliveryTag = response.getEnvelope().getDeliveryTag(); //获取消息的标识
                    channel.basicAck(deliveryTag, false);//拒绝一次取多条消息 一次只确认一条

                    byte[] msg = response.getBody();
                    log.info("从rabbitMQ中接收到的新订单消息", msg);
                    list.add(message);
                    ListUtil.reverse(list); //倒序
                    return list;
                }else {
                    log.info("接受新订单消息为空");
                    break;
                }
            }
        } catch (Exception e) {
            log.error("接受新订单失败", e);
            throw new GlobalException("接受新订单失败");
        }
        return null;
    }

    /**
     * 删除队列
     * @param userId
     */
    public void deleteNewOrderQueue(long userId) {
        String exchangeName = "new_order_private";
        String queueName = "queue_" + userId;
        try (
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel()
        ) {
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            channel.queueDelete(queueName);   //删除上面交换机中的队列

            log.info(userId + "删除新订单消息队列成功");
        } catch (Exception e) {
            log.error(userId + "删除新订单消息队列失败", e);
            throw new GlobalException("删除新订单消息队列失败");
        }
    }

    @Async
    public void deleteNewOrderQueueAsync(long userId) {
            this.deleteNewOrderQueue(userId);
    }

    /**
     * 清空队列中的消息
     * @param userId
     */
    public void clearNewOrderQueue(long userId) {
        String exchangeName = "new_order_private";
        String queueName = "queue_" + userId;
        try (
                Connection connection = factory.newConnection();
                Channel channel = connection.createChannel()
        ) {
            channel.exchangeDeclare(exchangeName, BuiltinExchangeType.DIRECT);
            channel.queuePurge(queueName);   //清空上面交换机中的这个队列

            log.info(userId + "清空新订单消息队列成功");
        } catch (Exception e) {
            log.error(userId + "清空新订单消息队列失败", e);
            throw new GlobalException("清空新订单消息队列失败");
        }
    }

    @Async
    public void clearNewOrderQueueAsync(long userId) {
        this.clearNewOrderQueue(userId);
    }
}
